package com.bbx.flink.demo.iterattions;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.stream.Stream;

public class IterativeDemo {
    public static void  main(String [] args){
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        IterativeStream<Long> iterate = env.generateSequence(0, 1000).iterate();
        SingleOutputStreamOperator<Long> minusOne = iterate.map(new MapFunction<Long, Long>() {
            private static final long serialVersionUID = -303565386342351751L;

            @Override
            public Long map(Long aLong) throws Exception {
                return aLong - 1;
            }
        });
        SingleOutputStreamOperator<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long aLong) throws Exception {
                return aLong > 0;
            }
        });
        iterate.closeWith(stillGreaterThanZero);
        SingleOutputStreamOperator<Long> filter = minusOne.filter(new FilterFunction<Long>() {
            private static final long serialVersionUID = 5866842237059970133L;

            @Override
            public boolean filter(Long aLong) throws Exception {
                boolean b = aLong <= 0;
                return b;
            }
        });
        

    }




}
